package com.tanghao.flink.demo.distinct;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.util.Collector;

/**
 * Created by Bill on 2019-5-9.
 */
public class MessageSplitter implements FlatMapFunction<String, Tuple4<Long, String, String, String>> {
    private Tuple4<Long, String, String, String> doaminFirstTime = new Tuple4<>(0L, "", "", "");
    @Override
    public void flatMap(String value, Collector<Tuple4<Long, String, String, String>> out) {
        if (value != null && value.contains(",")) {
            String[] parts = value.split(",");
            if(parts.length==9 && parts[6].length()!=0){
                doaminFirstTime.f0 = Long.parseLong(parts[1]);
                doaminFirstTime.f1 = parts[4];
                doaminFirstTime.f2 = parts[5];
                doaminFirstTime.f3 = parts[6];
                out.collect(doaminFirstTime);
            }
        }
    }
}